package chapter8.eg6;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.oio.OioEventLoopGroup;
import io.netty.channel.socket.DatagramPacket;
import io.netty.channel.socket.oio.OioDatagramChannel;
import io.netty.util.concurrent.Future;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
import org.junit.Test;

public class DatagramChannelExample {
  @Test
  public void testDatagramChannelClient() throws InterruptedException {
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(new OioEventLoopGroup())
        .channel(OioDatagramChannel.class)
        .handler(new ChannelInitializer<OioDatagramChannel>() {
          @Override
          protected void initChannel(OioDatagramChannel ch) throws Exception {
            ch.pipeline().addLast(new SimpleChannelInboundHandler<DatagramPacket>() {
              @Override
              protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg)
                  throws Exception {
                System.out.println("Receives data from server: " + msg.toString());
              }

              @Override
              public void channelActive(ChannelHandlerContext ctx) throws Exception {
                DatagramPacket msg = new DatagramPacket(
                    Unpooled.copiedBuffer("data from client".getBytes()), new InetSocketAddress(8080));
                ctx.writeAndFlush(msg);
              }
            });
          }
        });

    bootstrap.bind(8080).syncUninterruptibly();
    TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
  }

  @Test
  public void testDatagramChannelServer() throws InterruptedException {
    OioEventLoopGroup group = new OioEventLoopGroup();
    try {
      Bootstrap bootstrap = new Bootstrap();
      bootstrap.group(group)
          .channel(OioDatagramChannel.class)
          .handler(new SimpleChannelInboundHandler<DatagramPacket>() {
            private DatagramPacket packet;

            @Override
            protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg)
                throws Exception {
              System.out.println("Received data from server: " + msg.toString());
              packet = msg;
            }

            @Override
            public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
              DatagramPacket msg = new DatagramPacket(
                  Unpooled.copiedBuffer("data from server".getBytes()), packet.sender());
              ctx.writeAndFlush(msg);
            }
          });

      bootstrap.bind(8080).syncUninterruptibly();
    } finally {
      Future<?> future = group.shutdownGracefully();
      future.syncUninterruptibly();
    }

    TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
  }

}
